package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import groovy.time.TimeDuration;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        // 1. 从哪个主题获取流中的数据
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC);
        // 2. 定义流处理器 key: null    value: {article:111,type:LIKES,add:1}
        stream.map((key,value)->{
            log.info("流式处理  接收到行为数据:   {}",value);
            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
            //    key:  articleId     value:  type名称:add次数
            return new KeyValue<>(String.valueOf(updateArticleMess.getArticleId()),updateArticleMess.getType().name()+":"+updateArticleMess.getAdd());
        })
                .groupByKey()   // 根据key进行分组
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5))) // 时间窗口的设置， 每间隔5s  对该时间内的文章行为进行分组
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        // {"view":0,"like":0,"collection":0,"comment":0}
                        return JSON.toJSONString(new ArticleVisitStreamMess());
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     * @param key  分组内每一条数据的key
                     * @param value  分组内每一条数据的value   LIKES:1
                     * @param aggregate  上一次聚合的结果
                     * @return
                     */
                    @Override
                    public String apply(String key, String value, String aggregate) {
                        log.info("上一次聚合结果: {}  当前要计算的key: {}  ,value: {}",aggregate,key,value);
                        // {"view":0,"like":0,"collection":0,"comment":0}
                        ArticleVisitStreamMess preResult = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                        //  LIKES:1 ==>   ["LIKES", "1"]
                        String[] varArr = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(varArr[0])){
                            case LIKES:
                                // 设置 点赞数量
                                preResult.setLike(preResult.getLike() + Integer.parseInt(varArr[1]));
                                break;
                            case VIEWS:
                                // 设置 阅读数量
                                preResult.setView(preResult.getView() + Integer.parseInt(varArr[1]));
                                break;
                            case COMMENT:
                                // 设置 评论数量
                                preResult.setComment(preResult.getComment() + Integer.parseInt(varArr[1]));
                                break;
                            case COLLECTION:
                                // 设置 收藏数量
                                preResult.setCollect(preResult.getCollect() + Integer.parseInt(varArr[1]));
                                break;
                        }
                        return JSON.toJSONString(preResult);
                    }
                }).toStream()  // 将上面每个窗口的聚合结果  加入到流中
                .map((key,value)->{
                    // key {   key: 分组的key   , Window{  start     end }}
                    // value:  {"view":22,"like":11,"collection":33,"comment":44}
                    ArticleVisitStreamMess preResult = JSON.parseObject(value, ArticleVisitStreamMess.class);
                    preResult.setArticleId(Long.valueOf(key.key()));
                    log.info("时间窗口分组聚合结果: start:{}  end:{}  value:{}",
                            key.window().startTime().toString(),
                            key.window().endTime().toString(),
                            preResult);
                    return new KeyValue<>(key.key(),JSON.toJSONString(preResult));
                }).to(HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC);
        // 3. 将处理的结果发送到哪个主题
        return stream;
    }
}
